package cn.hexcloud.dga.governance.assessor.calc;

import cn.hexcloud.dga.common.util.SqlParser;
import cn.hexcloud.dga.ds.bean.TDsTaskDefinition;
import cn.hexcloud.dga.governance.assessor.Assessor;
import cn.hexcloud.dga.governance.bean.AssessParam;
import cn.hexcloud.dga.governance.bean.GovernanceAssessDetail;
import cn.hexcloud.dga.meta.bean.TableMetaInfo;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Sets;
import lombok.Getter;
import lombok.Setter;
import org.antlr.runtime.tree.Tree;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hive.ql.lib.Dispatcher;
import org.apache.hadoop.hive.ql.lib.Node;
import org.apache.hadoop.hive.ql.parse.ASTNode;
import org.apache.hadoop.hive.ql.parse.HiveParser;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.Stack;
import java.util.stream.Collectors;

@Component("SIMPLE_PROCESS")
public class SimpleProcessAssessor extends Assessor {
    @Override
    public void checkProblem(GovernanceAssessDetail governanceAssessDetail, AssessParam assessParam) throws Exception {
        //1    获取 sql    任务的定义
        TDsTaskDefinition tDsTaskDefinition = assessParam.getTDsTaskDefinition();
        if (tDsTaskDefinition == null) {
            return;
        }
        String sql = tDsTaskDefinition.getSql();
        if (sql == null) {
            return;
        }
        //2   构建一个节点处理器
        SimpleDispatcher simpleDispatcher = new SimpleDispatcher();
        simpleDispatcher.setDefaultSchemaName(assessParam.getTableMetaInfo().getSchemaName());
        //        2.1   尝试采集sql中的复杂操作包括： join  group by  union  function
        //         2.2  尝试采集 所有where 语句中涉及的字段
        //         2.3   采集被查询的表的名称 库名.表名
        //3   利用sql解析工具把sql 和处理器传入 ，遍历完成
        //获得 处理中已经采集的信息
        SqlParser.sqlParse(sql, simpleDispatcher);
        Set<String> complicateOperatorSet = simpleDispatcher.complicateOperatorSet;
        Set<String> fromTableSet = simpleDispatcher.fromTableSet;
        Set<String> whereFieldSet = simpleDispatcher.whereFieldSet;
        // 4    根据采集信息判断sql性质
        boolean isSimple = true;
        //        4.1  判断是否有复杂操作 ，只要有直接定性为复杂sql
        if (complicateOperatorSet.size() != 0) {
            isSimple = false;
        } else {
            //        4.2  如果没有复杂操作   获得被查询的表 通过元数据找到对应分区字段信息   用分区字段信息和 where涉及的字段进行比对  只要有一个字段不是分区字段，认定为复杂sql
            Set<String> names = new HashSet<>();
            for (TableMetaInfo tableMetaInfo : assessParam.getTableMetaInfoList()) {
                if (fromTableSet.contains(tableMetaInfo.getSchemaName() + "." + tableMetaInfo.getTableName())) {
                    continue;
                }
                List<JSONObject> jsonObjects = JSON.parseArray(tableMetaInfo.getPartitionColNameJson(), JSONObject.class);
                Set<String> name = jsonObjects.stream().map(a -> a.getString("name")).collect(Collectors.toSet());
                names.addAll(name);
            }
            if (!names.containsAll(whereFieldSet)) {
                isSimple = false;
            }
        }

        // 5    只要是简单sql就给0分
        if (isSimple) {
            governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
            governanceAssessDetail.setAssessProblem("简单加工");
        }
    }

    static class SimpleDispatcher implements Dispatcher {
        @Getter
        private Set<String> complicateOperatorSet = new HashSet<>();
        @Getter
        private Set<String> whereFieldSet = new HashSet<>();
        @Getter
        private Set<String> fromTableSet = new HashSet<>();
        @Setter
        private String defaultSchemaName = null;

        Set<Integer> complicateOperatorPatternSet = Sets.newHashSet(
                HiveParser.TOK_JOIN,  //join 包含通过where 连接的情况
                HiveParser.TOK_GROUPBY,       //  group by
                HiveParser.TOK_LEFTOUTERJOIN,       //  left join
                HiveParser.TOK_RIGHTOUTERJOIN,     //   right join
                HiveParser.TOK_FULLOUTERJOIN,     // full join
                HiveParser.TOK_FUNCTION,     //count(1)
                HiveParser.TOK_FUNCTIONDI,  //count(distinct xx)
                HiveParser.TOK_FUNCTIONSTAR, // count(*)
                HiveParser.TOK_SELECTDI,  // distinct
                HiveParser.TOK_UNIONALL   // union
        );

        @Override
        public Object dispatch(Node nd, Stack<Node> stack, Object... nodeOutputs) throws SemanticException {
            ASTNode astNode = (ASTNode) nd;
            // 尝试采集sql中的复杂操作
            if (complicateOperatorPatternSet.contains(astNode.getType())) {
                complicateOperatorSet.add(astNode.getText());
            }
            // 尝试采集所有where语句涉及的字段
            // 以HivePaese.TOK_TABLE_OR_COL 为基点，前提是该字段为WHERE下面而不是SELECT下面
            // 如果父级是DOT， 取父节点的第二个子节点；否则取自己的子节点
            if (astNode.getType() == HiveParser.TOK_TABLE_OR_COL && astNode.getAncestor(HiveParser.TOK_WHERE) != null) {
                Tree fieldNode = null;
                if (astNode.getParent().getType() == HiveParser.DOT) {
                    fieldNode = astNode.getParent().getChild(1);
                } else {
                    fieldNode = astNode.getChild(0);
                }
                whereFieldSet.add(fieldNode.getText());
            }
            // 采集被查询的表的名称 库名.表名
            // 以HiveParser.TOK_TABREF为基点，该节点的子节点长度为1时，则子节点为表名；为2时，为库名和表名
            if (astNode.getType() == HiveParser.TOK_TABREF) {
                ASTNode tableNameAllNode = (ASTNode) astNode.getChild(0);
                String schemaName = tableNameAllNode.getChildren().size() == 1 ? defaultSchemaName : tableNameAllNode.getChild(0).getText();
                String tableName = tableNameAllNode.getChildren().size() == 1 ? tableNameAllNode.getChild(0).getText() : tableNameAllNode.getChild(1).getText();
                fromTableSet.add(schemaName + "." + tableName);

            }
            return null;
        }
    }

    public static void main(String[] args) throws Exception {
        String sql1 = "insert overwrite table gmall.dwd_user_login_inc partition (dt = '${do_date}')\\nselect user_id,\\n       date_format(from_utc_timestamp(ts, 'GMT+8'), 'yyyy-MM-dd')          date_id,\\n       date_format(from_utc_timestamp(ts, 'GMT+8'), 'yyyy-MM-dd HH:mm:ss') login_time,\\n       channel,\\n       area_code                                                           province_id,\\n       version_code,\\n       mid_id,\\n       brand,\\n       model,\\n       operate_system\\nfrom (\\n         select user_id,\\n                channel,\\n                area_code,\\n                version_code,\\n                mid_id,\\n                brand,\\n                model,\\n                operate_system,\\n                ts\\n         from (\\n                  select user_id,\\n                         channel,\\n                         area_code,\\n                         version_code,\\n                         mid_id,\\n                         brand,\\n                         model,\\n                         operate_system,\\n                         ts,\\n                         row_number() over (partition by session_id order by ts) rn\\n                  from (select common.uid user_id,\\n                               common.ch  channel,\\n                               common.ar  area_code,\\n                               common.vc  version_code,\\n                               common.mid mid_id,\\n                               common.ba  brand,\\n                               common.md  model,\\n                               common.os  operate_system,\\n                               common.sid session_id,\\n                               ts\\n                        from gmall.ods_log_inc\\n                        where dt = '${do_date}'\\n                          and page is not null\\n                       ) t1\\n                  where a.user_id is not null\\n              ) t2\\n         where rn = 1\\n     ) t3";
        String sql2 = "insert overwrite table gmall.dwd_user_login_inc select a,b,c from abc where a in (select x from order_info oi where y='123')  ";
        sql1 = sql1.replace("\\n", " ");
        SimpleDispatcher simpleDispatcher = new SimpleDispatcher();
        simpleDispatcher.setDefaultSchemaName("gmall");

        SqlParser.sqlParse(sql1, simpleDispatcher);
        Set<String> fromTableSet = simpleDispatcher.getFromTableSet();
        Set<String> whereFieldSet = simpleDispatcher.getWhereFieldSet();
        Set<String> complicateOperatorSet = simpleDispatcher.getComplicateOperatorSet();
        System.out.println("StringUtils.join(fromTableSet,\",\") = " + StringUtils.join(fromTableSet, ","));
        System.out.println("StringUtils.join(whereFieldSet,\",\") = " + StringUtils.join(whereFieldSet, ","));
        System.out.println("StringUtils.join(complicateOperatorSet,\",\") = " + StringUtils.join(complicateOperatorSet, ","));
    }
}
